package com.hna.eking.SparkUtils;

import java.math.BigDecimal;
import java.sql.Date;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;



import scala.collection.Seq;

public class RowUtils implements Row {
	
	private ArrayList<RowContect> RowItem;
	private static StructType schema ;
	static final Logger logger = LoggerFactory.getLogger(RowUtils.class);
	int Pos;
	
	public RowUtils(){
		RowItem = new ArrayList();
		Pos = 0;
	}
	
	public void SetStructTypes(List<StructField> fields){
		schema = DataTypes.createStructType(fields);
	}
	
	public void AppendColume(String Name, Object Value) throws Exception{
		
//		if(schema == null){
//			logger.error("schema is null exception");
//			throw new Exception();
//		}
//		
//		int Pos = schema.fieldIndex(Name);
		
		RowContect row = new RowContect();
		
		if(Value == null){
			row.IsNull = true;
			row.value = null;
		}else{
			row.value = Value;
			row.IsNull = false;
		}
		row.ColumnName = Name;
		row.RowPos = Pos;
		RowItem.add(row);
		Pos++;		
	}
	
	public boolean anyNull() {
		// TODO Auto-generated method stub
		for(RowContect rw : RowItem){
			if(rw.IsNull == true)
				return true;
		}
		return false;
	}

	public Object apply(int idx) {
		// TODO Auto-generated method stub
		return RowItem.get(idx).value;
	}

	public Row copy() {
		// TODO Auto-generated method stub
		RowUtils ret = new RowUtils();
		ret.schema = this.schema();
		ret.RowItem = RowItem;
		return ret;
	}

	public int fieldIndex(String FieldName) {
		// TODO Auto-generated method stub
		for(RowContect rw : RowItem){
			if(rw.ColumnName.equals(FieldName))
				return rw.RowPos;
		}
		
		logger.error("Not Found Column Name In Row");
		//throw new Exception();
		return -1;
		
	}

	public Object get(int idx) {
		// TODO Auto-generated method stub
		
		return RowItem.get(idx).value;
	}

	public <T> T getAs(int idx) {
		// TODO Auto-generated method stub
		return (T) RowItem.get(idx).value;
	}

	public <T> T getAs(String FieldName) {
		// TODO Auto-generated method stub
		for(RowContect rw : RowItem){
			if(rw.ColumnName.equals(FieldName))
				return (T)rw.value;
		}
		logger.error("getAs<T> Not Found Column Name In Row");
		return null;
	}

	public boolean getBoolean(int idx) {
		// TODO Auto-generated method stub		
		return (Boolean) RowItem.get(idx).value;
	}

	public byte getByte(int idx) {
		// TODO Auto-generated method stub
		return (Byte) RowItem.get(idx).value;
	}

	public Date getDate(int idx) {
		// TODO Auto-generated method stub
		return (Date) RowItem.get(idx).value;
	}


	public BigDecimal getDecimal(int idx) {
		// TODO Auto-generated method stub
		return (BigDecimal) RowItem.get(idx).value;
	}


	public double getDouble(int idx) {
		// TODO Auto-generated method stub
		return (Double) RowItem.get(idx).value;
	}


	public float getFloat(int idx) {
		// TODO Auto-generated method stub
		return (Float) RowItem.get(idx).value;
	}


	public int getInt(int idx) {
		// TODO Auto-generated method stub
		return (Integer) RowItem.get(idx).value;
	}


	public <K, V> Map<K, V> getJavaMap(int idx) {
		// TODO Auto-generated method stub
		return null;
	}

	
	public <T> List<T> getList(int idx) {
		// TODO Auto-generated method stub
		return null;
	}


	public long getLong(int idx) {
		// TODO Auto-generated method stub
		return (Long) RowItem.get(idx).value;
	}

	public <K, V> scala.collection.Map<K, V> getMap(int idx) {
		// TODO Auto-generated method stub
		return (scala.collection.Map<K, V> ) RowItem.get(idx).value;
	}


	public <T> Seq<T> getSeq(int arg0) {
		// TODO Auto-generated method stub
		return null;
	}


	public short getShort(int idx) {
		// TODO Auto-generated method stub
		return (Short) RowItem.get(idx).value;
	}


	public String getString(int idx) {
		// TODO Auto-generated method stub
		return (String) RowItem.get(idx).value;
	}


	public Row getStruct(int idx) {
		// TODO Auto-generated method stub
		return (Row) RowItem.get(idx).value;
	}


	public Timestamp getTimestamp(int idx) {
		// TODO Auto-generated method stub
		return (Timestamp) RowItem.get(idx).value;
	}


	public <T> scala.collection.immutable.Map<String, T> getValuesMap(Seq<String> arg0) {
		// TODO Auto-generated method stub
		return null;
	}


	public boolean isNullAt(int idx) {
		return  RowItem.get(idx).IsNull;
	}


	public int length() {
		// TODO Auto-generated method stub
		return Pos;
	}


	public String mkString() {
		// TODO Auto-generated method stub
		String Ret = "";
		for(RowContect rw : RowItem){
			Ret = rw.value.toString() + ",";
		}
		int p = Ret.length();
		char[] arr = Ret.toCharArray();
		arr[p] = '\0';
		return new String(arr);
	}


	public String mkString(String FieldName) {
		// TODO Auto-generated method stub
		for(RowContect rw : RowItem){
			if(rw.ColumnName.equals(FieldName))
				return rw.value.toString();
		}
		return null;
	}


	public String mkString(String arg0, String arg1, String arg2) {
		// TODO Auto-generated method stub
		return null;
	}


	public StructType schema() {
		// TODO Auto-generated method stub
		return schema;
	}


	public int size() {
		// TODO Auto-generated method stub
		return Pos;
	}


	public Seq<Object> toSeq() {
		// TODO Auto-generated method stub
		//Seq sq = new Seq();
		
		return null;
	}

}
